package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo06WatermarkStrategy {
    public static void main(String[] args) throws Exception {
        // 事件时间：数据自带的时间(基于毫秒的时间戳)，Flink可以使用数据中的时间来确定是否触发窗口
        // 每5s统计一次，基于事件时间的滚动窗口

        /*
        准备数据：
        1744447972000 2025-04-12 16:52:52

        a,1744447972000
        a,1744447973000
        a,1744447974000
        a,1744447975000
        a,1744447976000
        a,1744447977000
        a,1744447978000
        a,1744447979000
        a,1744447981000
        a,1744447985000

        a,2025/04/12 16:52:51
        a,2025/04/12 16:52:52
        ......
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 将并行度设置为1更好的观察结果
        // 思考：如果并行度大于1，水位线如何计算？窗口如何触发？
        env.setParallelism(4);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);

        // 解析数据
        SingleOutputStreamOperator<Tuple2<String, Long>> kvDS = socketDS.map(line -> {
            String[] splits = line.split(",");
            String word = splits[0];
            long ts = Long.parseLong(splits[1]); // Flink只能识别以毫秒为单位的时间戳
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        // 设置时间戳以及水位线的策略
        SingleOutputStreamOperator<Tuple2<String, Long>> assignDS = kvDS.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {

            // 告诉Flink数据中的哪一部分是事件时间
            @Override
            public TimestampAssigner<Tuple2<String, Long>> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                return new TimestampAssigner<Tuple2<String, Long>>() {
                    // 从数据中提取时间戳
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                        return element.f1;
                    }
                };
            }

            // 指定水位线生成策略
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new MyWatermarkStrategy();
            }
        });

        assignDS
                // 将数据变成 word,1的KV格式
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                /*
                窗口的触发条件：水位线超过了窗口的结束时间
                 */
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                .print();


        env.execute();

    }
}

// 自定义的水位线策略
class MyWatermarkStrategy implements WatermarkGenerator<Tuple2<String, Long>> {

    private long maxTimestamp; // 保留接收到数据中最大的时间戳
    private final long outOfOrdernessMillis; // 容忍乱序的间隔时间

    public MyWatermarkStrategy() {
        this.outOfOrdernessMillis = 0;
    }

    public MyWatermarkStrategy(int seconds) {
        this.outOfOrdernessMillis = seconds * 1000L;
    }

    // 来一条数据执行一次
    @Override
    public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput output) {
        this.maxTimestamp = Math.max(this.maxTimestamp, event.f1); // 对比当前的maxTimestamp和数据中的时间戳，取最大值
        // 获取线程编号
        System.out.println("当前线程的ID：" + Thread.currentThread().getId());
        System.out.println("当前线程中最大的时间戳：" + this.maxTimestamp);
    }

    // 周期性地以200ms间隔时间emit出Watermark
    // 为何不是一条一条emit Watermark，性能太低
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(this.maxTimestamp - this.outOfOrdernessMillis));
    }
}